package org.example.flink.stream.demo;

import java.time.Duration;
import java.util.List;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class JoinDemo {
	
    public void run() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String dataSource = "[{'id':'1','username':'a'},{'id':'2','username':'b'},{'id':'3','username':'c'}]";
        List<JSONObject> list = JSONObject.parseArray(dataSource, JSONObject.class);
        DataStream<JSONObject> source = env.fromCollection(list);
        DataStream<JSONObject> source_=source.map(obj -> {
            obj.put("createdAt", System.currentTimeMillis());
            return obj;
        })
        .assignTimestampsAndWatermarks(WatermarkStrategy
        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(20))
        .withTimestampAssigner((event, timestamp) -> event.getLong("createdAt")));

        String dataSource2 = "[{'id':'1','age':'10'},{'id':'2','age':'15'},{'id':'3','age':'16'}]";
        List<JSONObject> list2 = JSONObject.parseArray(dataSource2, JSONObject.class);
        DataStream<JSONObject> source2 = env.fromCollection(list2);
        DataStream<JSONObject> source2_=source2.map(obj -> {
            obj.put("createdAt", System.currentTimeMillis());
            return obj;
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                .withTimestampAssigner((event, timestamp) -> event.getLong("createdAt")));
        

        DataStream<JSONObject> source3 = source_.join(source2_).where(obj -> {
            return obj.get("id");// 关联KEY
        }).equalTo(obj2 -> {
            return obj2.get("id");// 关联KEY
        })
        		.window(TumblingEventTimeWindows.of(Time.seconds(10)))
//        		.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(1)))
//        .window(EventTimeSessionWindows.withGap(Time.seconds(10)))
        .apply((obj1,obj2)->{
            	JSONObject newObj = new JSONObject();
                newObj.put("id", obj1.get("id"));
                newObj.put("username", obj1.get("username"));
                if (obj1.getString("id").equals(obj2.getString("id"))) {
                    newObj.put("age", obj2.get("age"));
                }
                return newObj;
            });
        source3.print();
        env.execute("TimeWindowDemo");
    }

    public static void main(String[] args) {
        try {
            new JoinDemo().run();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
